package com.ma.chapter8;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @ClassName TestPool
 * @Author: mayongqiang
 * @DATE 2022/3/20 13:37
 * @Description: 自定义线程池的实现！
 */
/*
 * 线程时一种系统资源,每创建一个线程就要占用一定的内存,
 * 1、高并发情况下创建多个线程，可能导致OOM
 * 2、创建多个线程,CPU负担大,线程上下文切换频繁,性能反而降低
 * 3、应该重复利用已有线程的潜力去处理任务(享元模式思想)
 * */
@Slf4j(topic = "m.TestPool")
public class MyPoolTest {
    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(1, 100,
                TimeUnit.MILLISECONDS, 1, (queue, task) -> {
            //1、死等
            queue.put(task);
            //2、带超时的等待
            //queue.offer(task, 500, TimeUnit.MILLISECONDS);
            //3、放弃   队列满剩余任务一个个放弃
            //log.debug("放弃 task {}",task);
            //4、抛异常  队列满不会往下继续执行
            //throw new RuntimeException("任务执行失败,task" + task);
            //5、调用者
            //task.run();
        });




        //当任务队列已满   继续等待/抛出异常   -->  加入拒绝策略
        /* 任务加入，队列已满 策略模式
                1、死等           queue.put(task);
                2、带超时的等待     queue.offer(task,timeout,timeUnit);
                3、让调用者放弃任务的执行 log.debug("放弃{}",task)
                4、让调用者抛出异常      throw new RuntimeException("任务执行失败"+task);
                5、让调用者自己完成任务
        * 这么多策略写死在ThreadPool中会造成很多elseif 代码臃肿
        * 权力下放，让调用者决定遇到这种情况（队列满）采取什么策略
        * 将决策权交给线程池的使用者，而不是写死在线程池中
        * --------------策略模式：将具体的操作抽象为接口，具体的实现由调用者传递进来
        */
        for (int i = 0; i < 5; i++) {
            int j = i;
            threadPool.execute(() -> {
                try {
                    Thread.sleep(5000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.debug("{}", j);
            });
        }
    }
}

/*
* 3、拒绝策略
    * 这么多策略写死在ThreadPool中会造成很多elseif 代码臃肿
    * 权力下放，让调用者决定遇到这种情况（队列满）采取什么策略
    * 将决策权交给线程池的使用者，而不是写死在线程池中
    * --------------策略模式：将具体的操作抽象为接口，具体的实现由调用者传递进来
     任务加入，队列已满 策略模式
        1、死等           taskQueue.put(task);
        2、带超时的等待     taskQueue.offer(task,timeout,timeUnit);
        3、让调用者放弃任务的执行 log.debug("放弃{}",task)
        4、让调用者抛出异常      throw new RuntimeException("任务执行失败"+task);
        5、让调用者自己完成任务
*/

//函数式接口，只包含一个抽象方法
@FunctionalInterface
interface RejectPolicy<T> {
    void reject(BlockingQueue<T> queue, T task);
}

/*
 * 2、自定义线程池
 * */
@Slf4j(topic = "m.ThreadPool")
class ThreadPool {
    //任务队列
    private BlockingQueue<Runnable> taskQueue;

    //线程集合
    private HashSet<Worker> workers = new HashSet<>();

    //核心线程数
    private int coreSize;

    //获取任务的超时时间
    private long timeout;

    private TimeUnit timeUnit;

    private RejectPolicy<Runnable> rejectPolicy;

    //执行任务
    public void execute(Runnable task) {
        //保证集合workers的线程安全性(临界区涉及 HashSet的size()和add()操作的组合！)
        synchronized (workers) {
            // 当任务数没有超过 coreSize 时,直接交给worker对象执行

            if (workers.size() < coreSize) {
                Worker worker = new Worker(task);
                log.debug("新增 worker{}, task {}", worker, task);
                workers.add(worker);
                worker.start();
            } else {
            // 当任务数超过 coreSize 时,加入任务队列暂存
                // 任务加入，队列已满 策略
                //1、死等           taskQueue.put(task);
                //2、带超时的等待     taskQueue.offer(task,timeout,timeUnit);
                //3、让调用者放弃任务的执行 log.debug("放弃{}",task)
                //4、让调用者抛出异常      throw new RuntimeException("任务执行失败"+task);
                //5、让调用者自己完成任务
                /*
                 * 这么多策略写死在ThreadPool中会造成很多elseif 代码臃肿
                 * 权力下放，让调用者决定遇到这种情况（队列满）采取什么策略
                 * 将决策权交给线程池的使用者，而不是写死在线程池中
                 * --------------策略模式：将具体的操作抽象为接口，具体的实现由调用者传递进来
                 * */
                taskQueue.tryPut(rejectPolicy, task);
            }
        }
    }

    public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapacity, RejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.rejectPolicy = rejectPolicy;
        this.taskQueue = new BlockingQueue<>(queueCapacity);
    }

    //对线程的包装,不断执行任务
    class Worker extends Thread {
        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            //执行任务  要么执行初始化时传过来的任务对象，要么从任务队列中取任务执行
            //1、当 task 不为空，执行任务
            //2、当 task 执行完了，再接着从任务队列中获取任务并执行
            /*
             * 策略1、不停等待任务 while (task != null || (task = taskQueue.take()) != null)
             * 策略2、超过超时时间回收线程 while(task != null||(task = taskQueue.poll(timeout,timeUnit)) != null)
             *   在 JDK线程池 中,分为 核心线程 和 最大线程
             *            --> 最大/救急线程用poll(销毁线程)    ==> 策略2
             *            --> 核心线程用take(归还线程池)  ==>策略1
             * */
            while (task != null || (task = taskQueue.take()) != null) {
            //while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
                try {
                    log.debug(" worker.run  正在执行 {}...", task);
                    task.run();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    task = null;
                }
            }
            synchronized (workers) {
                log.debug("worker run（） 被移除{}", this);
                workers.remove(this);
            }

        }
    }
}


/* 1、自定义任务队列->
 *      阻塞队列:生产者消费者模式下,平衡两者速度差异的组件
 *         线程池中的线程相当于任务的消费者,不断地去获取任务,执行任务
 *          该有任务的生产者线程,源源不断的产生新的任务
 *          这两者之间的速度差异,两种情况
 *                  没有任务(队列空) : 线程请求在阻塞队列等待(消费者在emptyCondition休息室等待)
 *                  任务很多(队列满) : 任务在阻塞队列等待(生产者在fullWaitCondition休息室等待)
 * */
@Slf4j(topic = "m.BlockingQueue")
class BlockingQueue<T> {
    //1、任务队列 用ArrayDeque双向队列实现
    private Deque<T> queue = new ArrayDeque<>();

    //2、锁 --> 保护队列头元素,避免被重复执行   保护队列尾元素,避免被重复添加
    private ReentrantLock lock = new ReentrantLock();

    //3、生产者条件变量 --> 一旦超过容量,避免生产者继续添加任务 队列满等待
    private Condition fullWaitCondition = lock.newCondition();

    //4、消费者条件变量  --> 队列空等待
    private Condition emptyCondition = lock.newCondition();

    //5、容量
    private int capacity;

    public BlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    //带超时的阻塞获取
    public T poll(long timeout, TimeUnit unit) {
        lock.lock();
        try {
            //将 timeout 同义转换为纳秒
            long nanos = unit.toNanos(timeout);
            //队列空 则等待emptyCondition被唤醒 (进入emptyCondition休息室等待)
            while (queue.isEmpty()) {
                try {
                    // 没有剩余时间,也即超时了-->不需要等待，直接返回
                    if (nanos <= 0) {
                        return null;
                    }
                    //返回值 --> 等待时间-已经经过的时间 = 剩余时间  也即还可以等待的时间
                    nanos = emptyCondition.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //队列不为空,取出队首的元素
            T t = queue.removeFirst();
            //唤醒生产者线程(叫醒在fullWaitCondition休息室中的等待的线程)
            fullWaitCondition.signal();
            return t;

        } finally {
            lock.unlock();
        }
    }


    //阻塞获取
    public T take() {
        lock.lock();
        try {
            //队列空 则等待emptyCondition被唤醒 (进入emptyCondition休息室等待)
            while (queue.isEmpty()) {
                try {
                    emptyCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //队列不为空,取出队首的元素
            T t = queue.removeFirst();
            //唤醒生产者线程(叫醒在fullWaitCondition休息室中的等待的线程)
            fullWaitCondition.signal();
            return t;

        } finally {
            lock.unlock();
        }
    }

    //带超时的阻塞添加
    public boolean offer(T task, long timeout, TimeUnit unit) {
        lock.lock();
        try {
            long nanos = unit.toNanos(timeout);
            //队列已满,等待fullWaitCondition被唤醒(进入fullWaitCondition休息室等待)
            while (queue.size() == capacity) {
                try {
                    if (nanos <= 0) {
                        return false;
                    }
                    log.debug("offer 等待加入队列 {}...", task);
                    nanos = fullWaitCondition.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //队列未满 队尾添加新元素
            log.debug("offer 加入任务队列{}", task);
            queue.addLast(task);
            //唤醒消费者线程(叫醒在emptyCondition休息室中等待的线程)
            emptyCondition.signal();
            return true;

        } finally {
            lock.unlock();
        }
    }

    //阻塞添加
    public void put(T task) {
        lock.lock();
        try {
            //队列已满,等待fullWaitCondition被唤醒(进入fullWaitCondition休息室等待)
            while (queue.size() == capacity) {
                try {
                    log.debug("put 等待加入任务队列 {}...", task);
                    fullWaitCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //队列未满 队尾添加新元素
            log.debug("put 加入任务队列 task-->{}", task);
            queue.addLast(task);
            //唤醒消费者线程(叫醒在emptyCondition休息室中等待的线程)
            emptyCondition.signal();

        } finally {
            lock.unlock();
        }

    }


    // 获取大小
    public int size() {
        lock.lock();
        try {
            return queue.size();
        } finally {
            lock.unlock();
        }
    }

    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
        lock.lock();
        try {
            //判断队列是否已满
            if (queue.size() == capacity) {
                //执行自定义拒绝策略
                rejectPolicy.reject(this, task);
            } else {
                //队列未满 队尾添加新元素
                log.debug("队列未满  tryPut 加入任务队列{}", task);
                queue.addLast(task);
                //唤醒消费者线程(叫醒在emptyCondition休息室中等待的线程)
                emptyCondition.signal();
            }
        } finally {
            lock.unlock();
        }
    }
}